<?php
/**
  * DGF
  *
  *
  * @copyright Copyright (c) 2012 - 2014, huxiaomin
  * @author huxiaomin    2012-5-13  
  * @since		Version 1.0
  *
  */

define ('CALLBACK_TIME',5);//callback失败重跑时间

class DTask{
    public $dao;
    public $table = 'dtask';
    public $table_log = 'dtask_log';
    public $dtask = '';
    public $dtask_log = '';
    
    public $callback_wait_time = 2;
    
    function __construct(){
        $this->dao = DDb::Dao();
        $this->dtask = new TTable('dtask');
        $this->dtask_log = new TTable('dtask_log');
    }
    
    public function start(){
    	$num = 20;//默认20个每次
        $task_arr = $this->select($num);
        foreach((array)$task_arr as $task_id => $task){
            //异步请求
            if ($task['type'] == 'request' && $task['status'] == 'new'){
                    $this->startRequest($task);
            }
            //异步回调
            else if ($task['type'] == 'request' && $task['status'] == 'back'){
            	    $this->startCallBack($task);
            }
            //异步回复
            else if ($task['type'] == 'response' && $task['status'] == 'new'){
                    $this->startResponse($task);	
            }
        }
    
    }
    
    
    
    /**
     * 异步请求
     * @param $task
     */
    public function startRequest($task,$time_out=5){
    	$task['type'] = 'request';
        $task['status'] = 'run';
        $id = $this->add($task);
        $task['id'] = $id;
        /*要求对方异步返回结果*/
        $input = $this->in('request',$task['request'],$task['id'],LOCAL_HOST.T_RECEIVE_HTTP);
        $return = TCurl::post($task['uri'],$input,$time_out);
        //失败重置任务
        if (!$return['success']){
        	$task['error'] = $return['error'];
            $task['status'] = 'new';
            $task['exec_nums']++;
            $seconds = $this->next($task['exec_nums']);
            $task['next_time'] = date('Y-m-d H:i:s', time() + $seconds);
            $this->add($task);
        }
        //成功访问并已得到数据
        if ($return['success'] == 1){
            if (!$task['callback']){
	            $task['end_time'] = date('Y-m-d H:i:s');
	            $task['status'] = 'close';
	            $task['exec_nums']++;
	            $this->add($task);
	            $this->finish($id);
	        }
	        $task['response'] = $return['data'];
	        $task['status'] = 'back';
	        $task['exec_nums']++;
	        $task['next_time'] = date('Y-m-d H:i:s', time() + 2);
	        $this->add($task);
        }
        return $return;
    }
    
    /**
     * 异步回复
     * @param $task
     */
    public function startResponse($task, $time_out=5){
    	$task['type'] = 'response';
        $task['status'] = 'run';
        $id = $this->add($task);
        $task['id'] = $id;
        //file_put_contents('thread.log.html',print_r($task,1),FILE_APPEND);
        
        $input = $this->in('response',$task['response'],$task['request'],0);
        $return = TCurl::post($task['uri'],$input,$time_out);
        //失败重置任务
        if (!$return['success']){
            $task['status'] = 'new';
            $task['error'] = $return['error'];
            $task['exec_nums']++;
            $seconds = $this->next($task['exec_nums']);
            $task['next_time'] = date('Y-m-d H:i:s', time() + $seconds);
            $this->add($task);
        }else {
            $task['end_time'] = date('Y-m-d H:i:s');
            $task['status'] = 'close';
            $task['exec_nums']++;
            $this->add($task);
            $this->finish($task['id']);
        }
        return $return;
    }
    
    /**
     * 异步回调
     * @param array $task
     * @param int $time_out
     */
    public function startCallBack($task, $time_out=30){
    	$task['type'] = 'request';
        $task['status'] = 'run';
        $id = $this->add($task);
        $task['id'] = $id;
        $input = $this->in('callback',$task['response'],$task['id'],0);
        /*只容许30秒*/
        $return_call = TCurl::post(LOCAL_HOST.T_CALLBACK_HTTP,$input,30);
        if ($return_call['success']){
            $task['error'] = $return_call['error'];
            $task['end_time'] = date('Y-m-d H:i:s');
            $task['status'] = 'close';
            $task['callback_data'] = $return_call['data'];
            $this->add($task);
            $this->finish($task['id']);
        }
        else {
        	//非超时失败，需重跑。
        	$task['error'] = $return_call['error'];
	        $task['status'] = 'back';
	        $task['exec_nums']++;
	        $seconds = $this->next($task['exec_nums']);
	        $task['next_time'] = date('Y-m-d H:i:s', time() + $seconds);
	        $this->add($task);
        }
        return $return_call;
    }
    
    
    /**
     * 添加任务，默认
     * @param $task
     */
    public function add($task){
    	if (isset($task['id']) && $task['id']){
    	   return $this->replace($task);
    	}
        $uniqid = mysql_real_escape_string((string)$task['uniqid']);
        if (!isset($task['last_id'])){
            $task['last_id'] = 0;
        }
        $last_id = (int)$task['last_id'];
        if (empty($last_id) && $uniqid){
            $sql = "SELECT max(id) FROM {$this->table} WHERE uniqid='{$uniqid}' ";
            $last_id = $this->dao->GetValue($sql);
        }       
        $task['last_id'] = $last_id;
        return $this->replace($task);
    }
    
    
    /**
     * 请求server 提供给外部使用
     * @param $task
     * @param $receive_async
     * @param $callback_async
     */
    public function add_sync($task,$receive_async=0, $callback_async=0, $receive_async_timeout=30){
        if (empty($task['uri'])){
            return $this->out(0,'','禁止无uri请求');
        }
    	$task['id'] = '';
        $task['type'] = 'request';
        $task['status'] = 'run'; // 设置任务正在运行，避免并发冲突
        $id = $this->add($task);
        $task['id'] = $id;
        /*实时获取接口数据*/
        if (!$receive_async){
           $input = $this->in('request',$task['request'],$id,0);
           $return = TCurl::post($task['uri'],$input,$receive_async_timeout);
           //失败直接抛弃
           if ($return['success'] !== 1){
           	    $task['error'] = $return['error'];
           	    $task['end_time'] = date('Y-m-d H:i:s');
                $this->add($task);
                $this->finish($id);
                return $this->out(0,'',$return['error']);     
           }
           //无回调的，结束返回就行。
           if (!$task['callback']){
           	    $task['end_time'] = date('Y-m-d H:i:s');
           	    $task['status'] = 'close';
           	    $this->add($task);
                $this->finish($id);
                return $this->out(1,$return['data']);
           }
           
           //成功后修改状态
           $task['response'] = $return['data'];
           
           //回调不异步的，
           if (!$callback_async){
           	    $return = $this->startCallBack($task);
           	    $return['data'] = $task['response'];
           	    return $return;
           }
           
           /*回调异步*/
           $task['status'] = 'back';
           $this->add($task);
           return $this->out(1,$task['response']);
           
        }
        return $this->startRequest($task,$receive_async_timeout);
    }
    
    
    
    /**
     * 标准返回
     * @param int $success
     * @param mix $data
     * @param string $error
     */
    public function out($success=0, $data='', $error=''){
        return array(
                     'success'=>$success,
                     'data'=>$data,
                     'error'=>$error
                    );
    }
    
    /**
     * 标准请求
     */
    public function in($type='', $request='', $task_id=0, $receive_async=0,$key=''){
        return array(
                    'type'=>$type,
                    'request'=>$request,
                    'receive_async'=>$receive_async,
                    'task_id'=>$task_id,
                    'key'=>$key,//TODO 账号密码安全保障
                    );
    }
    
    /**
     * 接收数据并调用回调
     */
    public function receive($in){
        //回调数据处理 返回值中有key response $response['response'];
        $task_id = $in['task_id'];
        if (empty($task_id)){
             return $this->out(0,'','task_id empty');       
        }
        $task = $this->getTask($task_id);
        if (!$task){
             return $this->out(0,'','not has row of task_id '.$task_id);   
        }
        if (!$task['callback']){
            $task['end_time'] = date('Y-m-d H:i:s');
            $task['status'] = 'close';
            $task['exec_nums']++;
            $this->add($task);
            $this->finish($task_id);
            return $this->out(1,'');
        }
        $task['response'] = $in['request'];
        $task['status'] = 'back';
        $task['exec_nums']++;
        $task['next_time'] = date('Y-m-d H:i:s', strtotime(time() + 2));
        $this->add($task);
        return $this->out(1,$task['response']);
    }
    
    
    /**
     * 根据id，进行insert或者update操作
     * @param array $add
     */
    function replace($add){
    	    if (!isset($add['id'])){
    	       $add['id'] = 0;
    	    }
    	    $id = $add['id'];
    	    if ($add['id']){
               $return = $this->dtask->update($add);
            }else {
               $return = $this->dtask->insert($add);
               $id = $return;
            }  	  
            if (!$return) {
            	//TODO log
            	return false;
            }
            return $id;
    }
    
    
  
    
    function next($num){
    	$num = (int)$num + 1;
    	if ($num > 24){
    	   return 2678400;//31天
    	}
    	else if ($num > 17){//1天
    	   return 86400;
    	}
    	else {
    	   return pow(2, $num);
    	}
    }
    
    
    /**
     * 任务移走
     * @param $id
     */
    function finish($id){
    	$sql = "UPDATE {$this->table} SET last_id=0,back_id={$id} WHERE last_id={$id}";
    	$this->dao->Query($sql);
        $sql = "INSERT INTO {$this->table_log} SELECT * FROM {$this->table} WHERE id={$id}";
        if ($this->dao->Query($sql)){
            $sql = "DELETE FROM {$this->table} WHERE id={$id} LIMIT 1";
            return $this->dao->Query($sql);
        }
        return false;
    }
    
    /**
     * 获取id值的任务
     * @param int $id
     * @return array 
     */
    function getTask($id){
    	$id = (int)$id;
        $sql = "SELECT * FROM {$this->table} WHERE id={$id}";
        return $this->dao->GetRow($sql);
    }
    
    function select($num = 20){
    	$now = date('Y-m-d H:i:s');
        $sql = "SELECT * FROM {$this->table} WHERE last_id=0 and status in ('new','back') and start_time<='{$now}' and next_time<='{$now}' limit {$num}";
        return $this->dao->GetPlan2($sql, 'id');
    }
    
    
    
    
}